package com.ldf.jrtsp.parser;

import com.ldf.jrtsp.callback.BareStreamCallBack;
import com.ldf.jrtsp.enums.RtpPayloadType;
import com.ldf.jrtsp.enums.RtspOptionsType;
import com.ldf.jrtsp.exception.JRtspException;
import com.sun.jna.Memory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import static com.ldf.jrtsp.constants.RtspConstants.RTSP_RTP_DATA_WAIT_TIME;

/**
 * rtp load PCMA数据解析
 *
 * @author lidaofu
 * @since 2024/6/21
 **/
public class RtpLoadPCMADataParser implements IRtpLoadCodecParser {
    private static final Logger log = LoggerFactory.getLogger(RtpLoadPCMADataParser.class);
    private final LinkedBlockingQueue<RtpDataParser> rtpDataQueue;
    private final BareStreamCallBack bareStreamCallBack;
    //为了和ZLM4J交互这里使用jna的memory 可以替换为nio的buffer 或者直接byte[]
    private Memory buffer;


    public RtpLoadPCMADataParser(BareStreamCallBack bareStreamCallBack) {
        this.rtpDataQueue = new LinkedBlockingQueue<>(256);
        this.bareStreamCallBack = bareStreamCallBack;
        this.buffer = new Memory(1024 * 1024);
    }

    /**
     * 发送rtp数据
     *
     * @param rtpDataParser
     */
    @Override
    public void sendRtpPacket(RtpDataParser rtpDataParser) {
        try {
            this.rtpDataQueue.offer(rtpDataParser, RTSP_RTP_DATA_WAIT_TIME, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            log.error("send rtp packet error queue full");
            throw new JRtspException("send rtp packet error queue full", e);
        }
    }

    @Override
    public void run() {
        while (true) {
            try {
                RtpDataParser rtpDataParser = this.rtpDataQueue.poll(RTSP_RTP_DATA_WAIT_TIME, TimeUnit.SECONDS);
                if (rtpDataParser == null) {
                    buffer.close();
                    break;
                }
                byte[] rawData = rtpDataParser.getRawData();
                buffer.write(0,rawData,0,rawData.length);
                bareStreamCallBack.callBack(RtpPayloadType.PCMA,buffer.share(0),rawData.length,rtpDataParser.getTimestamp());
            } catch (InterruptedException e) {
                log.error("take rtp packet error");
                throw new JRtspException("take rtp packet error", e);
            }
        }
    }
}